#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   For_Jenkins_2_test.py
@Time    :   2022/08/31 13:31:28
@Author  :   YangLiang 
@Version :   1.0
@Contact :   yangliang@chengtech.com
@License :   (C)Copyright 2022-2025
@Desc    :   None
'''

# here put the import lib
from pickle import FALSE, TRUE
import struct
from pytest_assume.plugin import assume
 
import pytest
import time
import requests
from sympy import I
# from sympy import assuming
from commDriver.TS_CAN import *
import allure
import sys
from collections.abc import Iterable

from sqlalchemy import false
sys.dont_write_bytecode = True

import project_CAN

from TimeSynchronization import *
import Set_up


class Test_Timesynchronization(object):

    def initialstate(self,Driver,iStep):
        iStep = Driver['CAN'].restartMRR_Power(Driver, iStep)
        Driver['CAN'].checkReqResp(Driver, "10 01", "50 01", 0, iStep, False)
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: front time synchronization check  0x66F byte[4] #  byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr)) 
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        
        return iStep


    @allure.severity('normal')
    @allure.title('Test do not time synchronization in default session with physical addressing')
    @allure.description('Test do not time synchronization in default session with physical addressing')
    def test_Without_Timesynchronization_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)
       

    @allure.severity('normal')
    @allure.title('Test time synchronization in default session with physical addressing')
    @allure.description('Test time synchronization in default session with physical addressing')
    def test_Timesynchronization_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver)

        # 等待同步帧接收
        time.sleep(0.1)

        # 读取时间同步帧内容
        for i in range(len(Set_up.MRR_0x66F_data)):
            respData = Set_up.MRR_0x66F_data[i]
            if respData[4] == 0x08:
                readtime = Set_up.MRR_0x66F_time[i]
                break
        # 打印帧内容  
        actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
        logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
        logging.info('sendtime = [{0}]ms, rxtime = [{1}]'.format(tl.sendtime,readtime)) 

        # 判断时间同步状态
        with allure.step(f'Step{iStep}: time synchronization success check 0x66F byte[4] # 0x66F byte[4] == 0x08'):
            with assume: assert respData[4] == 0x08
            iStep = iStep + 1

        # 判断时间帧误差
        with allure.step(f'Step{iStep}: time synchronization success check  0x66F timestamp #  |timestamp - timeNumber| < 8'):
            btimestamp = respData[5:11]
            f = int.from_bytes(bytes(btimestamp), byteorder='little')
            logging.info('write_timestamp=[{0}]ms, read_timestamp=[{1}]ms'.format(tl.timeNumber*1000, f))
            f = f- tl.timeNumber*1000
            f1 = readtime - tl.sendtime
            logging.info('Frame_Time_difference = [{0}]ms, CAN_Time_difference = [{1}]'.format(f,f1))
            with assume: assert abs(f-f1) < 8
            iStep = iStep + 1

        tl.exit = True


    @allure.severity('normal')
    @allure.title('Test time synchronization with out sync in default session with physical addressing')
    @allure.description('Test time synchronization with out sync in default session with physical addressing')
    def test_Timesynchronization_NoSync_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver,skipSync=True)
        Set_up.MRR_0x66F_data.clear()
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: time synchronization with out sync check  0x66F byte[4] # 0x66F byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True

    @allure.severity('normal')
    @allure.title('Test time synchronization with out fup in default session with physical addressing')
    @allure.description('Test time synchronization with out fup in default session with physical addressing')
    def test_Timesynchronization_NoFup_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver,skipFup=True)
        Set_up.MRR_0x66F_data.clear()
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: time synchronization with out fup check  0x66F byte[4] # 0x66F byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True


    @allure.severity('normal')
    @allure.title('Test time synchronization with out 5s in default session with physical addressing')
    @allure.description('Test time synchronization with out 5s in default session with physical addressing')
    def test_Timesynchronization_TimeOut_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)
        
        # 开始时间同步
        tl = TimeSynchron(Driver)
        time.sleep(1)
        tl.exit = True

        time.sleep(5.5)
        with allure.step(f'Step{iStep}: time synchronization with out 5s check  0x66F byte[4] # 0x66F byte[4] == 0x01'):
            Set_up.MRR_0x66F_data.clear()
            time.sleep(0.1)
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x01
            iStep = iStep + 1

    @allure.severity('normal')
    @allure.title('Test Front time synchronization with out fup 0.3s in default session with physical addressing')
    @allure.description('Test Front time synchronization with out fup 0.3s in default session with physical addressing')
    def test_FrontTimesynchronization_FupTimeOut_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver, FupWaitTime = 0.2)
        
        with allure.step(f'Step{iStep}: time synchronization with out fup 0.3s check  0x66F byte[4] # 0x66F byte[4]  == 0x00'):
            Set_up.MRR_0x66F_data.clear()
            time.sleep(0.1)
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True

    @allure.severity('normal')
    @allure.title('Test time After synchronization with out fup 0.3s in default session with physical addressing')
    @allure.description('Test After time synchronization with out fup 0.3s in default session with physical addressing')
    def test_AfterTimesynchronization_FupTimeOut_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver)
        time.sleep(1)
        tl.exit = True
        tl.join()
        time.sleep(5.5)
        # 开始时间同步
        tl = TimeSynchron(Driver, FupWaitTime = 0.2)
        
        with allure.step(f'Step{iStep}: time synchronization with out fup 0.3s check  0x66F byte[4] # 0x66F byte[4]  == 0x00'):
            Set_up.MRR_0x66F_data.clear()
            time.sleep(0.1)
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x01
            iStep = iStep + 1
        tl.exit = True


    @allure.severity('normal')
    @allure.title('Test repeate time synchronization with now time in default session with physical addressing')
    @allure.description('Test repeate time synchronization with now time in default session with physical addressing')
    def test_Repeate_Timesynchronization_OnNowTime_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver)

        # 等待同步帧接收

        for j in range(10):
            Set_up.MRR_0x66F_data.clear()
            Set_up.MRR_0x66F_time.clear()
            time.sleep(0.1)
            # 读取时间同步帧内容
            for i in range(len(Set_up.MRR_0x66F_data)):
                respData = Set_up.MRR_0x66F_data[i]
                if respData[4] == 0x08:
                    readtime = Set_up.MRR_0x66F_time[i]
                    break
            # 打印帧内容  
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            logging.info('sendtime = [{0}]ms, rxtime = [{1}]'.format(tl.sendtime,readtime)) 

            # 判断时间同步状态
            with allure.step(f'Step{iStep}: time synchronization success check 0x66F byte[4] # 0x66F byte[4] == 0x08'):
                with assume: assert respData[4] == 0x08
                iStep = iStep + 1

            # 判断时间帧误差
            with allure.step(f'Step{iStep}: time synchronization success check  0x66F timestamp #  |timestamp - timeNumber| < 8'):
                btimestamp = respData[5:11]
                f = int.from_bytes(bytes(btimestamp), byteorder='little')
                logging.info('write_timestamp=[{0}]ms, read_timestamp=[{1}]ms'.format(tl.timeNumber*1000, f))
                f = f- tl.timeNumber*1000
                f1 = readtime - tl.sendtime
                logging.info('Frame_Time_difference = [{0}]ms, CAN_Time_difference = [{1}]'.format(f,f1))
                with assume: assert abs(f-f1) < 8
                iStep = iStep + 1

        tl.exit = True

    @allure.severity('normal')
    @allure.title('Test repeate time synchronization with defined time in default session with physical addressing')
    @allure.description('Test repeate time synchronization with defined time in default session with physical addressing')
    def test_Repeate_Timesynchronization_OndefinedTime_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        
        # 等待同步帧接收
        for j in range(10):
            tl = TimeSynchron(Driver,definedTime=True,definedIndex=j)
            Set_up.MRR_0x66F_data.clear()
            Set_up.MRR_0x66F_time.clear()
            time.sleep(0.2)
            # 读取时间同步帧内容
            # for i in range(len(Set_up.MRR_0x66F_data)):
            #     respData = Set_up.MRR_0x66F_data[i]
            #     if respData[4] == 0x08:
            #         readtime = Set_up.MRR_0x66F_time[i]
            #         respData = Set_up.MRR_0x66F_data[i]
            #         break
            i = len(Set_up.MRR_0x66F_data) -1
            readtime = Set_up.MRR_0x66F_time[i]
            respData = Set_up.MRR_0x66F_data[i]

            # 打印帧内容  
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            logging.info('sendtime = [{0}]ms, rxtime = [{1}]'.format(tl.sendtime,readtime)) 

            # 判断时间同步状态
            with allure.step(f'Step{iStep}: time synchronization success check 0x66F byte[4] # 0x66F byte[4] == 0x08'):
                with assume: assert respData[4] == 0x08
                iStep = iStep + 1

            # 判断时间帧误差
            with allure.step(f'Step{iStep}: time synchronization success check  0x66F timestamp #  |timestamp - timeNumber| < 8'):
                btimestamp = respData[5:11]
                f = int.from_bytes(bytes(btimestamp), byteorder='little')
                logging.info('write_timestamp=[{0}]ms, read_timestamp=[{1}]ms'.format(tl.timeNumber*1000, f))
                f = f- tl.timeNumber*1000
                f1 = readtime - tl.sendtime
                logging.info('Frame_Time_difference = [{0}]ms, CAN_Time_difference = [{1}]'.format(f,f1))
                with assume: assert abs(f-f1) < 8
                iStep = iStep + 1
            tl.exit = True
            tl.join()

        tl.exit = True

    @allure.severity('normal')
    @allure.title('Test time synchronization with All CRC failed in default session with physical addressing')
    @allure.description('Test time synchronization with All CRC failed in default session with physical addressing')
    def test_Timesynchronization_AllCrcFailed_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver, Sync_crcFailed=True,Fup_crcFailed=True)
        Set_up.MRR_0x66F_data.clear()
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: time synchronization with out sync check  0x66F byte[4] # 0x66F byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True


    @allure.severity('normal')
    @allure.title('Test time synchronization with Sync CRC failed in default session with physical addressing')
    @allure.description('Test time synchronization with Sync CRC failed in default session with physical addressing')
    def test_Timesynchronization_SyncCrcFailed_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver, Sync_crcFailed=True)
        Set_up.MRR_0x66F_data.clear()
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: time synchronization with out sync check  0x66F byte[4] # 0x66F byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True

    
    @allure.severity('normal')
    @allure.title('Test time synchronization with Fup CRC failed in default session with physical addressing')
    @allure.description('Test time synchronization with Fup CRC failed in default session with physical addressing')
    def test_Timesynchronization_FupCrcFailed_Default_phy(self, action, Driver):
        iStep = 1
        iStep = self.initialstate(Driver,iStep)

        # 开始时间同步
        tl = TimeSynchron(Driver, Fup_crcFailed=True)
        Set_up.MRR_0x66F_data.clear()
        time.sleep(0.1)
        with allure.step(f'Step{iStep}: time synchronization with out sync check  0x66F byte[4] # 0x66F byte[4] == 0x00'):
            respData = Set_up.MRR_0x66F_data[len(Set_up.MRR_0x66F_data)-1]
            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            with assume: assert respData[4] == 0x00
            iStep = iStep + 1
        tl.exit = True

    
    

    
    @allure.severity('normal')
    @allure.title('Test time synchronization Precision in default session with physical addressing')
    @allure.description('Test time synchronization in default session with physical addressing')
    def test_Timesynchronization_Precision_Default_phy(self, action, Driver):
        iStep = 1
        diffTime = []
        
        # 时间同步成功
        self.test_Timesynchronization_Default_phy(action, Driver)

        with allure.step(f"Step{iStep}:只发送2F6 Sync报文"):
            logging.info('只发送2F6 Sync报文')

        for i in range(20):
            tl = TimeSynchron(Driver)
            tl.exit = True
            Driver['CAN2'].fush_q_buff()  # 清缓存  
            time.sleep(0.1)
            ALl_data,buffer_size = Driver['CAN2'].read_frame()
            response_data = ALl_data[ALl_data['AIdentifier'] == 0x66F]
            for j in range(len(response_data)):
                if response_data.iloc[j]['ADatas'][4] == 0x08:
                    respData = response_data.iloc[j]['ADatas'][:16]
                    break
            btimestamp = respData[5:11]
            lotime  = int.from_bytes(bytes(btimestamp), byteorder='little')

            actuRespStr = binascii.b2a_hex(bytearray(respData), ' ')    
            logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x66F, actuRespStr))
            logging.info(' 发送0x2F6时间同步报文后:0x66F报文回复的时间为 :{}'.format(lotime))
            logging.info(' 时间同步发送数据:{}'.format(tl.timeNumber*1000))
            with assume: assert abs(lotime-tl.timeNumber*1000) <100     #比对请求回复
            result=(lotime-tl.timeNumber*1000)
            if result <100:
                logging.info('目标同步数据在100ms内')
            else:
                logging.error('目标同步数据在100ms外')
            logging.info('目标同步数据差值为：{}'.format(result))
            diffTime.append(result)
            time.sleep(1)

        logging.info('{3}次时间差值:最大值=[{0}], 最小值=[{1}], 均值=[{2}]'.format(max(diffTime), min(diffTime),  sum(diffTime) / len(diffTime), i + 1))
        
        

        


    
